package com.chukun.disruptor.higher.multi;

import com.chukun.disruptor.higher.chain.Trade;
import com.lmax.disruptor.*;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

/**
 * @author chukun
 * disruptor多消费者，多生产者模式
 */
public class DisruptorMultiPatternMainRunner {

    public static void main(String[] args) throws InterruptedException {

        int ringBufferSize = 1024*1024;
        //创建RingBuffer实例
        RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        },ringBufferSize,new YieldingWaitStrategy());

        //通过ringBuffer创建屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        //创建消费者数组
         TradeConsumer[] consumers = new TradeConsumer[10];
         for (int i=0;i<consumers.length;i++){
             consumers[i]=new TradeConsumer("consumer-"+i);
         }
         //构建多消费者工作池
        WorkerPool<Trade> workerPool = new WorkerPool<>(ringBuffer,sequenceBarrier,new TradeEventExceptionHandler(),consumers);
         //设置多个消费者的sequence序号，用于单独统计消费进度，并且设置到ringBuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        //启动workerPool
        workerPool.start(Executors.newFixedThreadPool(5));

        final CountDownLatch latch = new CountDownLatch(1);
        for(int i=0;i<100;i++){
            TradeProducer producer = new TradeProducer(ringBuffer);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    for(int j=0;j<100;j++){
                        producer.sendData(UUID.randomUUID().toString());
                    }
                }
            }).start();
        }
        Thread.sleep(2000);//构建线程
        System.err.println("----------线程创建完毕，开始生产数据----------");

        latch.countDown();
        Thread.sleep(10000);
        System.err.println("任务总数:" + consumers[2].getCount());

    }
}

/**
 * 异常处理的类
 */
class TradeEventExceptionHandler implements ExceptionHandler<Trade>{

    @Override
    public void handleEventException(Throwable throwable, long l, Trade trade) {

    }

    @Override
    public void handleOnStartException(Throwable throwable) {

    }

    @Override
    public void handleOnShutdownException(Throwable throwable) {

    }
}
